
package com.dremio.exec.store.jdbc.conf;

import com.dremio.exec.catalog.conf.*;
import com.dremio.exec.store.jdbc.CloseableDataSource;
import com.dremio.exec.store.jdbc.DataSources;
import com.dremio.exec.store.jdbc.JdbcPluginConfig;
import com.dremio.exec.store.jdbc.dialect.OceanBaseMysqlDialect;
import com.dremio.exec.store.jdbc.dialect.arp.ArpDialect;
import com.dremio.exec.store.jdbc.legacy.JdbcDremioSqlDialect;
import com.dremio.exec.store.jdbc.legacy.LegacyCapableJdbcConf;
import com.dremio.exec.store.jdbc.legacy.LegacyDialect;
import com.dremio.exec.store.jdbc.legacy.MySQLLegacyDialect;
import com.dremio.options.OptionManager;
import com.dremio.security.CredentialsService;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import io.protostuff.Tag;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.reflect.MethodUtils;

import javax.sql.ConnectionPoolDataSource;
import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotBlank;
import java.lang.reflect.InvocationTargetException;
import java.sql.SQLException;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;

@SourceType(
        value = "OCEANBASE-MYSQL",
        label = "OCEANBASE-MYSQL",
        uiConfig = "oceanbase-layout.json"
)
public class OceanBaseMysqlConf extends AbstractArpConf<OceanBaseMysqlConf> {
    private static final String ARP_FILENAME = "arp/implementation/oceanbase-mysql-arp.yaml";
//    private static final OceanBaseMysqlDialect oceanBaseMysqlDialect =
//            AbstractArpConf.loadArpFile(ARP_FILENAME, OceanBaseMysqlDialect::new);

    private static final ArpDialect oceanBaseMysqlDialect = AbstractArpConf.loadArpFile(ARP_FILENAME, (ArpDialect::new));
    private static final String DRIVER = "com.mysql.cj.jdbc.Driver";
    @Tag(1)
    @DisplayMetadata(
            label = "Host"
    )
    public @NotBlank String hostname;
    @Tag(2)
    @DisplayMetadata(
            label = "Port"
    )
    public @NotBlank @Min(1L) @Max(65535L) String port = "2881";

    @Tag(3)
    @DisplayMetadata(label = "Database")
    public String database;

    @Tag(4)
    public String username;
    @Tag(5)
    @Secret
    public String password;
    @Tag(6)
    public AuthenticationType authenticationType;
    @Tag(7)
    @DisplayMetadata(
            label = "Record fetch size"
    )
    @NotMetadataImpacting
    public int fetchSize = 200;
    @Tag(8)
    @DisplayMetadata(
            label = "Net write timeout (in seconds)"
    )
    @NotMetadataImpacting
    public int netWriteTimeout = 60;
//    @Tag(9)
//    @DisplayMetadata(
//            label = "Enable legacy dialect"
//    )
//    public boolean useLegacyDialect = false;
    @Tag(10)
    @NotMetadataImpacting
    @DisplayMetadata(
            label = "Grant External Query access (Warning: External Query allows users with the Can Query privilege on this source to query any table or view within the source)"
    )
    public boolean enableExternalQuery = false;

    @Tag(11)
    public List<Property> propertyList;

    @Tag(12)
    @DisplayMetadata(
            label = "Enable legacy dialect"
    )
    public boolean compact = false;

    public OceanBaseMysqlConf() {
    }

    @Override
    public JdbcDremioSqlDialect getDialect() {
        return oceanBaseMysqlDialect ;
    }

    @VisibleForTesting
    public JdbcPluginConfig buildPluginConfig(JdbcPluginConfig.Builder configBuilder, CredentialsService credentialsService, OptionManager optionManager) {
        return configBuilder.withDialect(this.getDialect()).
                withDatasourceFactory(this::newDataSource).
                withShowOnlyConnDatabase(false).
                withFetchSize(this.fetchSize).
                withAllowExternalQuery(this.enableExternalQuery).
                build();
    }

    private CloseableDataSource newDataSource() throws SQLException {
        Properties properties = new Properties();
        if (null != this.propertyList) {
            this.propertyList.forEach((p) -> {
                properties.put(p.name, p.value);
            });
        }
        //source = (ConnectionPoolDataSource)Class.forName(POOLED_DATASOURCE).newInstance();
        return DataSources.newGenericConnectionPoolDataSource(DRIVER,
                toJdbcConnectionString(),
                username,
                password,
                properties,
                DataSources.CommitMode.FORCE_MANUAL_COMMIT_MODE);

    }

    @VisibleForTesting
    CloseableDataSource newDataSource(ConnectionPoolDataSource source) throws SQLException {
        String url = this.toJdbcConnectionString();
        Map<String, String> properties = new LinkedHashMap<>();
        properties.put("useJDBCCompliantTimezoneShift", "true");
        properties.put("sessionVariables", String.format("net_write_timeout=%d", this.netWriteTimeout));
        String encodedProperties = (String) properties.entrySet().stream().map((entry) -> {
            return (String) entry.getKey() + '=' + (String) entry.getValue();
        }).collect(Collectors.joining("&"));

        try {
            MethodUtils.invokeExactMethod(source, "setUrl", new Object[]{url});
            if (this.username != null) {
                MethodUtils.invokeExactMethod(source, "setUser", new Object[]{this.username});
            }

            if (this.password != null) {
                MethodUtils.invokeExactMethod(source, "setPassword", new Object[]{this.password});
            }

            MethodUtils.invokeExactMethod(source, "setProperties", new Object[]{encodedProperties});
            return DataSources.newSharedDataSource(source);
        } catch (InvocationTargetException var7) {
            Throwable cause = var7.getCause();
            if (cause != null) {
                Throwables.throwIfInstanceOf(cause, SQLException.class);
            }

            throw new RuntimeException("Cannot instantiate MySQL datasource", var7);
        } catch (ReflectiveOperationException var8) {
            throw new RuntimeException("Cannot instantiate MySQL datasource", var8);
        }
    }

    private String toJdbcConnectionString() {
        String hostname = (String) Preconditions.checkNotNull(this.hostname, "missing hostname");
        String portAsString = (String) Preconditions.checkNotNull(this.port, "missing port");
        int port = Integer.parseInt(portAsString);
        if (StringUtils.isEmpty(this.database)){
            return String.format("jdbc:mysql://%s:%d", hostname, port);
        }else {
            return String.format("jdbc:mysql://%s:%d/%s", hostname, port, database);
        }
    }

//    protected LegacyDialect getLegacyDialect() {
//        return MySQLLegacyDialect.INSTANCE;
//    }

    protected ArpDialect getArpDialect() {
        return oceanBaseMysqlDialect;
    }

    @VisibleForTesting
    public static ArpDialect getDialectSingleton() {
        return oceanBaseMysqlDialect;
    }

}
